package com.hanson.demo.actors;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * LinkedBlockingQueue 多用于任务队列（单线程发布任务，任务满了就停止等待阻塞，当任务被完成消费少了又开始负载 发布任务）
 * ConcurrentLinkedQueue 多用于消息队列（多个线程发送消息，先随便发来，不计并发的-cas特点） 单生产者，单消费者 用
 * LinkedBlockingqueue 多生产者，单消费者 用 LinkedBlockingqueue 单生产者 ，多消费者 用
 * ConcurrentLinkedQueue 多生产者 ，多消费者 用 ConcurrentLinkedQueue
 *
 */
@Data
@Slf4j
public class ActorMailbox implements ActorCtx {

	// @Value("${actors.system.throughput:5}")
	private int actorThroughput = 5;
	// 改用非阻塞
	private final ConcurrentLinkedQueue<Message> mailboxQueue = new ConcurrentLinkedQueue<>();
	private ExecutorService executorService = Executors.newFixedThreadPool(2);
	private final Actor actor;
	private final String selfId;
	private final Dispatcher dispatcher;
	// 消息状态
	private static final boolean FREE = false;
	private static final boolean BUSY = true;
	private static final boolean NOT_READY = false;
	private static final boolean READY = true;
	private final AtomicBoolean busy = new AtomicBoolean(FREE);
	private final AtomicBoolean ready = new AtomicBoolean(NOT_READY);
	private final AtomicBoolean destroyInProgress = new AtomicBoolean(); // 销毁

	public void startActor() {
		dispatcher.getExecutor().execute(() -> tryInit(1));
	}

	private void tryInit(int attempt) {
		try {
			if (!destroyInProgress.get()) {
				ready.set(READY);// 开始状态
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void processMailbox() {
		boolean noMoreElements = false;
		for (int i = 0; i < actorThroughput; i++) {
			Message message = mailboxQueue.poll(); // 移除并返回队列头部的元素 如果队列为空，则返回null‌
			if (message != null) {
				actor.process(message);
			} else {
				noMoreElements = true;
				break;
			}
		}
		if (noMoreElements) {
			busy.set(FREE);
			dispatcher.getExecutor().execute(() -> tryProcessQueue(false));
		} else {
			dispatcher.getExecutor().execute(this::processMailbox);
		}

	}

	@Override
	public void tell(Message actorMsg) {
		enqueue(actorMsg);
	}

	private void enqueue(Message msg) {
		mailboxQueue.add(msg);
		tryProcessQueue(true);
	}

	private void tryProcessQueue(boolean newMsg) {
		if (ready.get() == READY) {
			if (newMsg || !mailboxQueue.isEmpty()) {
				if (busy.compareAndSet(FREE, BUSY)) {
					dispatcher.getExecutor().execute(this::processMailbox);// 快速处理Mailbox消息
				} else {
					//log.trace("[{}] MessageBox is busy, new msg: {}", selfId, newMsg);
					System.err.println("MessageBox is busy");
				}
			}
		} else {
			log.trace("[{}] MessageBox is not ready, new msg: {}", selfId, newMsg);
		}
	}

	@Override
	public void stop() {
		ready.set(NOT_READY);// 状态
		executorService.shutdownNow();
	}

	@Override
	public String getSelf() {
		return selfId;
	}

}
